package com.wyw.springcloud.biz;

import com.wyw.springcloud.topic.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * 文件介绍
 *
 * @Author： Mr Wu （yewen.wu.china.gmail.com）
 * @Date: 2021/3/9 17:42
 */
@Slf4j
@EnableBinding(value = {
        Sink.class,
        MyTopic.class,
        GroupTopic.class,
        DelayedTopic.class,
        ErrorTopic.class,
        RequeueTopic.class,
        DlqTopic.class,
        FallbackTopic.class
})
public class StreamConsumer {

    private AtomicInteger count = new AtomicInteger(1);

    @StreamListener(Sink.INPUT)//将信道绑定到方法上
    public void consume(Object payload){
        log.info("接收到消息，payload={}", payload);

    }

    @StreamListener(MyTopic.INPUT)//将信道绑定到自定义的topic上
    public void myConsume(Object payload){
        log.info("my Consumer接收到消息，payload={}", payload);

    }

    @StreamListener(GroupTopic.INPUT)//将信道绑定到自定义的轮询topic上
    public void groupTopicConsume(Object payload){
        log.info("groupTopic接收到消息，payload={}", payload);

    }

    @StreamListener(DelayedTopic.INPUT)//将信道绑定到自定义的轮询topic上
    public void groupDelayedTopicConsume(MessageBean messageBean){
        log.info("groupTopic接收到消息，payload={}", messageBean.getPayload());

    }

    @StreamListener(ErrorTopic.INPUT)//将信道绑定到自定义的轮询topic上
    public void groupErrorTopicConsume(MessageBean messageBean){
        log.info("Are you OK?");

        if (count.incrementAndGet() % 3 == 0) {
            log.info("Fine, thank you. And you?");
            count.set(0);
        } else {
            log.info("What's your problem?");
            throw new RuntimeException("I'm not OK");
        }

    }

    @StreamListener(RequeueTopic.INPUT)//将信道绑定到自定义的轮询topic上
    public void groupRequeueTopicConsume(MessageBean messageBean){
        log.info("Are you OK?");
        try {
            Thread.sleep(3 * 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        throw new RuntimeException("I'm not OK");

    }

    //死信队列
    @StreamListener(DlqTopic.INPUT)//将信道绑定到自定义的轮询topic上
    public void groupDlqTopicConsume(MessageBean messageBean){
        log.info("dlq - Are you OK?");

        if (count.incrementAndGet() % 3 == 0) {
            log.info("dlq - Fine, thank you. And you?");
            count.set(0);
        } else {
            log.info("dlq - What's your problem?");
            throw new RuntimeException("I'm not OK");
        }

    }

    //降级队列
    @StreamListener(DlqTopic.INPUT)//将信道绑定到自定义的轮询topic上
    public void groupFallbackTopicConsume(MessageBean messageBean,
                                          @Header("version") String version){
        log.info("fallback - Are you OK?");

        if ("1.0".equals(version)) {
            log.info("fallback - Fine, thank you. And you?");
        } else if ("2.0".equals(version)) {
            log.info("fallback - 不支持此版本");
            throw new RuntimeException("I'm not OK");
        }  else {
            log.info("fallback - version:{}", version);
        }

    }

    //指定输入信道
    @ServiceActivator(inputChannel = "fallback-topic.fallback-group.errors")
    public void fallback(Message<?> message){
        log.info("fallback entered");

    }
}
